package com.citi.kafka.serializer.customer;

import com.sun.xml.internal.ws.encoding.soap.SerializationException;
import org.apache.kafka.common.protocol.types.Field;
import org.apache.kafka.common.serialization.Deserializer;
import org.apache.kafka.common.serialization.Serializer;

import java.nio.ByteBuffer;
import java.util.Map;

public class UserDeSerializer implements Deserializer<User> {

    @Override
    public void configure(Map<String, ?> map, boolean b) {

    }

    @Override
    public User deserialize(String topic, byte[] data) {
        if(data==null)
            return null;
        if(data.length<12)
            throw new SerializationException("the Person data bytes length should not be less than 12");
        ByteBuffer buffer = ByteBuffer.wrap(data);
        int id = buffer.getInt();
        //parse the name
        int nameLenght = buffer.getInt();
        byte[] nameBytes = new byte[nameLenght];
        buffer.get(nameBytes);
        String name=new String(nameBytes);

        //parse the address
        int addrLength = buffer.getInt();
        byte[] addrBytes = new byte[addrLength];
        buffer.get(addrBytes);
        String addr=new String(addrBytes);



        return new User(id,name,addr);
    }

    @Override
    public void close() {

    }
}
